package ntd.util.frame.rmq;

import cn.hutool.core.thread.ThreadUtil;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.log.StaticLog;
import com.rabbitmq.client.*;
import lombok.AllArgsConstructor;
import lombok.Data;
import ntd.util.frame.rmq.bo.ContextRmq;
import ntd.util.frame.rmq.bo.RmqMsgBO;

import java.io.IOException;
import java.util.concurrent.TimeoutException;
import java.util.function.Consumer;

@Data
@AllArgsConstructor
public class Rmq {

    private Connection connection;
    private ContextRmq contextRmq;

    public static Rmq init(ContextRmq contextRmq) throws IOException, TimeoutException {
        //1.创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();//MQ采用工厂模式来完成连接的创建
        //2.在工厂对象中设置连接信息(ip,port,virtualhost,username,password)
        factory.setHost(contextRmq.getHost());//设置MQ安装的服务器ip地址
        factory.setPort(contextRmq.getPort());//设置端口号
        factory.setVirtualHost(contextRmq.getVirtualHost());//设置虚拟主机名称
        //MQ通过用户来管理
        factory.setUsername(contextRmq.getUsername());//设置用户名称
        factory.setPassword(contextRmq.getPassword());//设置用户密码
        //3.通过工厂对象获取连接
        Connection connection = factory.newConnection();
        StaticLog.info("成功连接到RMQ服务-[{}]", contextRmq.getHost());
        return new Rmq(connection, contextRmq);
    }

    public void pubMsg(String exchange, String routingKey, String msg) throws IOException, TimeoutException {
        //mq提供Channel来将处理消息
        //创建Channel
        Channel channel = connection.createChannel();
        //basicPublish将消息发送到指定的交换机
        channel.basicPublish(exchange, routingKey, null, msg.getBytes());
        //关闭连接
        channel.close();
    }

    public void subMsg(String exchange, String routingKey, Consumer<RmqMsgBO> consumer) throws IOException {
        //创建Channel
        Channel channel = connection.createChannel();
        // 临时队列
        String queue = channel.queueDeclare().getQueue();
        //基于route key 绑定队列和交换机
        channel.queueBind(queue, exchange, routingKey);
        //通过basicConsumer方法从指定队列中获取消息，消息生产者会通过ex2交换机中的key值将消息发送到queue6中，因为在创建queue6时绑定到交换机ex3中，指定的路由键为a。
        //consumer参数是消息接收之后的处理方法
        channel.basicConsume(queue,true, new DefaultConsumer(channel){
            //重写handleDelivery方法（这个方法是消息的处理过程）
            //body参数就是接受到的消息
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) {
                //将消息转换成String类型然后打印
                StaticLog.info("收到RMQ消息[{}]", new String(body));
                if (ObjectUtil.isNull(consumer)) {
                    return;
                }
                consumer.accept(new RmqMsgBO(consumerTag, envelope, properties, body));
            }
        });
        StaticLog.info("RMQ开始监听-exchange[{}]-routingKey[{}]", exchange, routingKey);
    }

    public static void main(String[] args) throws IOException, TimeoutException {
        ContextRmq contextRmq = new ContextRmq();
        contextRmq.setHost("112.13.217.188");
        contextRmq.setPort(6005);
        contextRmq.setUsername("root");
        contextRmq.setPassword("t5cJJCte");
        Rmq rmq = Rmq.init(contextRmq);
        sub(rmq);
        rmq.pubMsg("hik.municipal.device.exchange", "hik.municipal.device.alarm.routingkey", "{}");
    }

    private static void sub(Rmq rmq) throws IOException {
        ThreadUtil.execAsync(() -> {
            try {
                rmq.subMsg("hik.municipal.device.exchange", "hik.municipal.device.alarm.routingkey", null);
            } catch (IOException e) {
                StaticLog.error(e, "[海康][预警平台][{}]失败...");
            }
        });
        rmq.subMsg("hik.municipal.device.exchange", "hik.municipal.device.data.routingkey", null);
        StaticLog.info("执行完成");
    }


}
